package com.gxlevi.task

import com.gxlevi.bean.ClickLogWide
import com.gxlevi.util.HBaseUtil
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.sink.SinkFunction

case class ChannelRealHot(var channelID: String, var visited: Long)

object ChannelRealHotTask extends BaseTask[ChannelRealHot] {
  override def map(clickWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelRealHot] = {
    clickWideDataStream.map {
      clickLog =>
        ChannelRealHot(clickLog.channelID, clickLog.count)
    }
  }

  override def keyBy(mapStream: DataStream[ChannelRealHot]): KeyedStream[ChannelRealHot, String] = {
    mapStream.keyBy {
      line => line.channelID
    }
  }

  override def reduce(windowedStream: WindowedStream[ChannelRealHot, String, TimeWindow]): DataStream[ChannelRealHot] = {
    windowedStream.reduce {
      (t1, t2) =>
        ChannelRealHot(t1.channelID, t1.visited + t2.visited)
    }
  }

  override def sinkToHBase(reduceDataStream: DataStream[ChannelRealHot]): Unit = {
    reduceDataStream.addSink(new SinkFunction[ChannelRealHot] {
      override def invoke(value: ChannelRealHot): Unit = {
        tableName = "channel"
        clfName = "info"
        val visitedColName = "visited"
        rowKey = value.channelID

        val visitedValue = HBaseUtil.getMapData(tableName, rowKey, clfName, List(visitedColName))

        val totalVisited = getTotal(visitedValue, visitedColName, value.visited)
        HBaseUtil.putMapData(tableName, rowKey, clfName, Map(
          channelIdColumn -> value.channelID,
          visitedColName -> totalVisited
        ))
      }
    })
  }
}
